package org.keystate

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：key state example类
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/24 20:24
 * modified By ：
 * version:    : 1.0
 */
object KeyedStateExample1 extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 操作
    source
      .keyBy(_.id)
      .flatMap(new TemperatureAlertFunction(2))
      .print
  }

  class TemperatureAlertFunction(d: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
    // 记录上一次温度
    lazy val lastTmp = getRuntimeContext.getState(
      new ValueStateDescriptor[Double]("lastTmp", Types.of[Double])
    )

    override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]) = {
      val temperature = value.temperature
      val lastTempValue = lastTmp.value()
      if (lastTempValue != 0.0) { // 第一次不计算
        // 获取本次温度和上一次温度差值
        if ((lastTempValue - temperature).abs > d) {
          out.collect((value.id, lastTempValue, temperature))
        }
      }
      lastTmp.update(temperature)
    }
  }

}
